package rpc

import (
	"bytes"
	"cluster-cache/store"
	"fmt"
	"github.com/panjf2000/gnet"
	"github.com/panjf2000/gnet/pkg/logging"
	"sync/atomic"
	"time"
)

type TcpServerV1 struct {
	*gnet.EventServer
	eng          gnet.Server
	network      string
	addr         string
	connected    int32
	disconnected int32
	handler      Handler
}

const BufferSize8k = 8 * 1024

func NewTcpServerV1(addr string, hand Handler) *TcpServerV1 {
	return &TcpServerV1{
		network:      "tcp",
		addr:         addr,
		connected:    0,
		disconnected: 0,
		handler:      hand,
	}
}

func (s *TcpServerV1) OnInitComplete(srv gnet.Server) (action gnet.Action) {
	logging.Infof("running server on %s", fmt.Sprintf("%s://%s", s.network, s.addr))
	s.eng = srv
	return
}

func (s *TcpServerV1) OnOpened(srv gnet.Conn) (out []byte, action gnet.Action) {
	code := HeadBodyCodecPool.Get().(*HeadBodyCodec)
	code.Reset()
	srv.SetContext(code)
	atomic.AddInt32(&s.connected, 1)
	return
}

func (s *TcpServerV1) OnClosed(c gnet.Conn, err error) (action gnet.Action) {
	if err != nil {
		logging.Infof("error occurred on connection=%s, %v\n", c.RemoteAddr().String(), err)
	}
	atomic.AddInt32(&s.connected, -1)
	HeadBodyCodecPool.Put(c.Context())
	return
}

func (s *TcpServerV1) React(frame []byte, c gnet.Conn) (out []byte, action gnet.Action) {
	code := c.Context().(*HeadBodyCodec)
	if len(frame) == 0 {
		return
	}
	reader := bytes.NewReader(frame)
	futures := make([]CallFuture, 0)
	for {
		data, seq, err := code.Decode(reader)
		if err != nil {
			if err == ErrIncompletePacket {
				break
			}
			code.Reset()
			break
		}
		future := s.handler.Accept(data)
		if future != nil {
			futures = append(futures, future...)
		}
		code.Reset()
		packet, _ := EncodeWithSeq(store.EmptyByte, seq)
		out = packet
	}
	for i := range futures {
		futures[i].WaitDone()
	}
	return
}

func (s *TcpServerV1) Start() {
	err := gnet.Serve(s, s.addr,
		gnet.WithMulticore(true),
		gnet.WithTCPKeepAlive(time.Minute*5),
		gnet.WithReusePort(true),
		gnet.WithSocketRecvBuffer(BufferSize8k),
		gnet.WithSocketSendBuffer(BufferSize8k),
		gnet.WithTCPNoDelay(gnet.TCPNoDelay),
	)
	if err != nil {
		logging.Errorf("start server error", err)
	}
}
